package com.tsf.hbase.jpa.dao;

import com.tsf.hbase.HBaseClient;
import com.tsf.hbase.HBaseClientFactory;
import com.tsf.hbase.HBaseTemplate;
import com.tsf.hbase.config.HBaseProperties;
import com.tsf.hbase.jpa.HBaseColumn;
import com.tsf.hbase.jpa.HBaseRepository;
import com.tsf.hbase.jpa.HBaseTable;
import com.tsf.hbase.jpa.Page;
import com.tsf.hbase.jpa.entity.HBaseEntity;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.annotation.AnnotationConfigurationException;

import javax.annotation.PostConstruct;
import java.lang.reflect.Field;
import java.lang.reflect.ParameterizedType;
import java.util.*;

/**
 * <p>‘hbase’DAO层的抽象实现，实现一些自动扫描，自动注入，自动对象转换等反射功能</p>
 * <h2>更多抽象实现：</h2>
 * <ul><li>请查看{@link BaseHBaseDao}的操作选项实现</li></ul>
 *
 * @param <T> hbase实体类的基本定义，基于{@code BaseHBaseDao}的操作类所指定的
 *            实体对象必须继承至{@link HBaseEntity}，因为这样才能让{@code AbstractHBaseDao}
 *            正确加载，否则会导致服务启动失败
 * @author taoningbo
 * @since 1.0, Created as 2018/11/26
 */
public abstract class AbstractHBaseDao<T extends HBaseEntity> {

    protected final Logger log = LoggerFactory.getLogger(this.getClass());

    private final Class<T> entityClass;
    private final HBaseTable hBaseTable;
    final Map<HBaseColumn, Field> columns = new HashMap<>();
    protected String tableName;
    protected Collection<String> families;

    @Autowired
    protected HBaseTemplate hBaseTemplate;

    @Autowired
    private HBaseProperties properties;

    @Autowired
    private HBaseClientFactory factory;

    /**
     * 构造一个抽象操作实例，只有当子类被加载时才会被触发此方法。
     * <p>
     * 1、完成自动加载数据表的表名，列族名，表的属性等配置；
     * 2、完成{@link HBaseTemplate}、{@link HBaseProperties}和{@link HBaseClientFactory}的注入。
     * </p>
     */
    protected AbstractHBaseDao() {
        @SuppressWarnings("unchecked")
        Class<T> entityClass = (Class<T>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
        if (entityClass == null) {
            throw new AnnotationConfigurationException(
                    "The correct 'hbase' entity object is not specified, and a subclass of 'HBaseEntity' should be specified.");
        }
        this.entityClass = entityClass;

        HBaseTable hBaseTable = entityClass.getAnnotationsByType(HBaseTable.class)[0];
        if (hBaseTable == null) {
            throw new AnnotationConfigurationException(String.format(
                    "The HBaseEntity '%s' no implement HBaseTable.", entityClass.getName()));
        }
        this.hBaseTable = hBaseTable;

        if (StringUtils.isEmpty(hBaseTable.name())) {
            throw new IllegalArgumentException(
                    "The 'name' attribute in 'HBaseTable' must be specified and cannot be an empty or an empty string.");
        }
        this.tableName = hBaseTable.name();

        if (hBaseTable.families().length <= 0) {
            throw new IllegalArgumentException(
                    "The 'families' attribute in 'HBaseTable' must be specified and cannot be an empty or an empty string.");
        }
        this.families = Arrays.asList(hBaseTable.families());
    }

    /**
     * 自动创建数据表并加载基于{@link HBaseEntity}的所有实体类属性。
     */
    @PostConstruct
    public final void autoCreateHTable() {
        if (properties == null || factory == null || hBaseTemplate == null) {
            throw new IllegalArgumentException("The 'spring.hbase.enabled' configuration is not enabled. If you" +
                    " want to use it this way, please enable automatic initialization and at least fill in the" +
                    " basic configuration in 'HBaseProperties'.");
        }
        factory.checkInitialized();

        // 如果配置文件中未指定数据表属性，则注解中的优先，否则以配置中的属性优先
        if (properties.getTables().isEmpty()) {
            HBaseProperties.Table tableConf = new HBaseProperties.Table();
            tableConf.setTableName(hBaseTable.name());
            tableConf.setFamilyNames(hBaseTable.families());
            tableConf.setCompactionEnabled(hBaseTable.compactionEnabled());
            tableConf.setCompression(hBaseTable.compressionType());
            tableConf.setBloomFilterType(hBaseTable.bloomFilterType());
            tableConf.setBlockCache(hBaseTable.blockCache());
            tableConf.setBlockSize(hBaseTable.blockSize());
            tableConf.setTimeToLive(hBaseTable.timeToLive());
            if (StringUtils.isNotEmpty(hBaseTable.splitKeys())) {
                tableConf.setSplitKeys(hBaseTable.splitKeys());
            }
            factory.autoCreateTable(tableConf);
            log.info("The hbase data table is being automatically created using the table properties specified by the annotations in the code.");
        } else {
            log.warn("You are using the hbaseJpa schema and setting the native 'Table' configuration, the system will use the native 'Table' configuration first!");
        }

        // 加载每个实体类中的所有字段
        for (Field field : entityClass.getDeclaredFields()) {
            HBaseColumn[] columns = field.getAnnotationsByType(HBaseColumn.class);
            if (columns.length != 0) {
                HBaseColumn column = columns[0];
                if (StringUtils.isEmpty(column.family())) {
                    throw new IllegalArgumentException("The family name to which the column belongs is not set.");
                }
                if (!families.contains(column.family())) {
                    throw new IllegalArgumentException(String.format(
                            "The family name specified by the column '%s' does not exist in '@HBaseTable$families()'!",
                            column.family()));
                }
                this.columns.put(column, field);
            }
        }
        log.info("{} class load done, tableName:{}|families:{}", this.getClass().getCanonicalName(), hBaseTable.name(), hBaseTable.families());
    }

    /**
     * 检查hbase-dao是否有标注‘HBaseRepository’注解类
     *
     * @param daoClass 当前类
     * @return 如果有标注返回‘true’，否则返回‘false’
     */
    protected static boolean checkedRepository(Class<?> daoClass) {
        HBaseRepository[] repositories = daoClass.getAnnotationsByType(HBaseRepository.class);
        return repositories.length == 1;
    }

    /**
     * 构建一个{@link Put}的对象
     *
     * @param entity 实体对象
     * @return 返回构建好的{@link Put}实例
     * @throws IllegalAccessException 如果当获取实体中的某个字段的值出错时，才将抛出此错误
     */
    protected final Put buildPut(T entity) throws IllegalAccessException {
        if (entity.rowKey() == null) {
            throw new IllegalAccessException("未设置");
        }
        Put put = null;
        for (HBaseColumn column : columns.keySet()) {
            Field field = columns.get(column);
            field.setAccessible(true);
            byte[] value = getFieldValue(entity, field);
            if (put == null) {
                put = new Put(entity.rowKey());
            }
            put.addColumn(Bytes.toBytes(column.family()), Bytes.toBytes(column.name()), value);
        }
        return put;
    }

    /**
     * 将hbase的结果{@link Result}转换{@link HBaseEntity}类型的对象
     *
     * @param result 数据结果
     * @return 返回一个实体类型的结果
     */
    protected final T parseResult(Result result) {
        try {
            T entity = entityClass.newInstance();
            entity.setRowKey(result.getRow());
            for (HBaseColumn column : columns.keySet()) {
                Field field = columns.get(column);
                field.setAccessible(true);
                byte[] value = result.getValue(Bytes.toBytes(column.family()), Bytes.toBytes(column.name()));
                if (value != null && value.length > 0) {
                    setFieldValue(entity, field, value);
                }
            }
            return entity;
        } catch (Exception e) {
            log.error("Parse entity result error.", e);
        }
        return null;
    }

    /**
     * 转换一批{@link ResultScanner}结果为{@link T}对象的集合，当{@code page}有值时需要
     * 设置下一次访问的起始{@code rowKey}。
     *
     * @param results 待转换的一批结果数据
     * @param page    分页对象（当此值为空时，表示不分页使用）
     * @return 返回结果实例的集合
     */
    protected final Collection<T> parseResults(ResultScanner results, Page page) {
        ArrayList<T> entities = new ArrayList<>();
        for (Result result : results) {
            if (!result.isEmpty()) {
                T parseResult;
                if ((parseResult = parseResult(result)) != null) {
                    entities.add(parseResult);
                }
            }
        }
        if (page != null) {
            int lastIndex = entities.size() - 1;
            if (lastIndex == page.getPageSize()) {
                T entity = entities.remove(lastIndex);
                page.setFirstRowKey(entity.rowKeyAsString());
            }
        }
        return entities;
    }

    /**
     * 根据字段的类型获取字段中的值
     *
     * @param entity 实体对象
     * @param field  字段
     * @return 返回一个hbase的bytes类型的字段值
     * @throws IllegalAccessException 如果获取实体中的值存在错误时会抛出前者错误或后者错误，如果你实体中字段类型不在我们规定之中，也将会抛出后者错误
     */
    protected final byte[] getFieldValue(T entity, Field field) throws IllegalAccessException {
        byte[] value;
        Class<?> fieldType = field.getType();
        if (fieldType.getTypeName().equals(String.class.getTypeName())) {
            value = field.get(entity) == null ? null : Bytes.toBytes(String.valueOf(field.get(entity)));
        } else if (fieldType.getTypeName().equals(Integer.class.getTypeName()) || fieldType.getTypeName().equals("int")) {
            value = field.get(entity) == null ? null : Bytes.toBytes((int) field.get(entity));
        } else if (fieldType.getTypeName().equals(Float.class.getTypeName()) || fieldType.getTypeName().equals("float")) {
            value = field.get(entity) == null ? null : Bytes.toBytes((float) field.get(entity));
        } else if (fieldType.getTypeName().equals(Double.class.getTypeName()) || fieldType.getTypeName().equals("double")) {
            value = field.get(entity) == null ? null : Bytes.toBytes((double) field.get(entity));
        } else if (fieldType.getTypeName().equals(Boolean.class.getTypeName()) || fieldType.getTypeName().equals("boolean")) {
            value = field.get(entity) == null ? null : Bytes.toBytes((boolean) field.get(entity));
        } else if (fieldType.getTypeName().equals(Long.class.getTypeName()) || fieldType.getTypeName().equals("long")) {
            value = field.get(entity) == null ? null : Bytes.toBytes((long) field.get(entity));
        } else if (fieldType.getTypeName().equals(Short.class.getTypeName()) || fieldType.getTypeName().equals("short")) {
            value = field.get(entity) == null ? null : Bytes.toBytes((short) field.get(entity));
        } else if (fieldType.getTypeName().equals(byte[].class.getTypeName())) {
            value = field.get(entity) == null ? null : (byte[]) field.get(entity);
        } else {
            throw new IllegalArgumentException("Field type error, not matched to legal field type!");
        }
        return value;
    }

    /**
     * 给指定的字段{@code field}赋值
     *
     * @param entity 实体对象
     * @param field  字段
     * @param value  字段对应的值
     * @throws IllegalAccessException 如果在设置字段值的过程中存在错误，将会抛出此错误
     */
    protected final void setFieldValue(T entity, Field field, byte[] value) throws IllegalAccessException {
        Class<?> fieldType = field.getType();
        if (fieldType.getTypeName().equals(String.class.getTypeName())) {
            field.set(entity, Bytes.toString(value));
        } else if (fieldType.getTypeName().equals(Integer.class.getTypeName()) || fieldType.getTypeName().equals("int")) {
            field.set(entity, Bytes.toInt(value));
        } else if (fieldType.getTypeName().equals(Float.class.getTypeName()) || fieldType.getTypeName().equals("float")) {
            field.set(entity, Bytes.toFloat(value));
        } else if (fieldType.getTypeName().equals(Double.class.getTypeName()) || fieldType.getTypeName().equals("double")) {
            field.set(entity, Bytes.toDouble(value));
        } else if (fieldType.getTypeName().equals(Boolean.class.getTypeName()) || fieldType.getTypeName().equals("boolean")) {
            field.set(entity, Bytes.toBoolean(value));
        } else if (fieldType.getTypeName().equals(Long.class.getTypeName()) || fieldType.getTypeName().equals("long")) {
            field.set(entity, Bytes.toLong(value));
        } else if (fieldType.getTypeName().equals(Short.class.getTypeName()) || fieldType.getTypeName().equals("short")) {
            field.set(entity, Bytes.toShort(value));
        } else if (fieldType.getTypeName().equals(byte[].class.getTypeName())) {
            field.set(entity, value);
        }
    }

    /**
     * 获取指定实体字段名{@code fieldName}的所属族名称
     *
     * @param fieldName 实体字段名
     * @return 返回一个bytes类型的列族名称
     */
    protected final byte[] getFamily(String fieldName) {
        byte[] family = null;
        for (HBaseColumn column : columns.keySet()) {
            Field field = columns.get(column);
            if (field.getName().equals(fieldName)) {
                family = Bytes.toBytes(column.family());
                break;
            }
        }
        return family;
    }

    /**
     * 获取指定实体字段名{@code fieldName}的对应列限定符的名称
     *
     * @param fieldName 实体字段名
     * @return 返回一个bytes类型的列限定符的名称
     */
    protected final byte[] getQualifier(String fieldName) {
        byte[] qualifier = null;
        for (HBaseColumn column : columns.keySet()) {
            Field field = columns.get(column);
            if (field.getName().equals(fieldName)) {
                qualifier = Bytes.toBytes(column.name());
                break;
            }
        }
        return qualifier;
    }

    /**
     * 获取hbase的客户端对象
     *
     * @return 返回初始化的hbase客户端
     */
    public abstract HBaseClient getClient();

    /**
     * 将实体中传的值转换为过滤条件列表。
     *
     * @param entity 实体对象
     * @return FilterList
     */
    protected final FilterList getFilterList(T entity) {
        FilterList filters = new FilterList();
        for (HBaseColumn column : columns.keySet()) {
            Field field = columns.get(column);
            field.setAccessible(true);
            byte[] value;
            try {
                value = getFieldValue(entity, field);
            } catch (IllegalAccessException e) {
                log.warn("The value of the field '{}' cannot be obtained.", field.getName());
                continue;
            }
            if (value != null) {
                SingleColumnValueFilter filter = new SingleColumnValueFilter(
                        Bytes.toBytes(column.family()),
                        Bytes.toBytes(column.name()),
                        CompareFilter.CompareOp.EQUAL,
                        value);
                filters.addFilter(filter);
            }
        }
        return filters;
    }
}
